package com.qisi.utils;

import com.qisi.pojo.AppType;
import com.qisi.pojo.alertManager.Route;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ConfigOptions;
import org.apache.flink.configuration.Configuration;

import java.io.IOException;
import java.io.Serializable;
import java.time.LocalDateTime;
import java.util.Collections;
import java.util.List;


@Slf4j
public class Config extends Configuration implements Serializable {

    private static Configuration config;
    // APP 类型，决定了处理日志的类型
    public static AppType APP_TYPE ;

    //public static final List<String> KAFKA_CONSUMER_TOPIC_LIST = Collections.singletonList("events");
    public static List<String> KAFKA_CONSUMER_TOPIC_LIST ;
    public static String KAFKA_PRODUCER_TOPIC = "sentry-sink-topic-test";
    public static String KAFKA_PRODUCER_CONFIG_FILE = "alert/producer.properties";
    public static String KAFKA_CONSUMER_CONFIG_FILE = "alert/kafka.properties";


    // 车机端埋点日志告警内容的 key
    public static final String KAFKA_LOG_VIN_KEY = "common_data.vin";
    public static final String KAFKA_LOG_ALERT_NAME_KEY = "common_data.appPackage";
    public static final String KAFKA_LOG_APPLICATION_ID_KEY = "common_data.traceId";
    public static final String KAFKA_LOG_BEHAVIOR_ID_KEY = "common_data.behaviorId";
    // 云端日志告警内容的 key {"stackTrace": 70868, "traceId": 29946, "logTime": "2021-06-30 14:45:42", "appName": 88489, "serviceInstanceIp": 85385, "level": "error", "vin": 84360, "appVersion": 24768, "version": "1.1", "deviceId": 39073, "uniqueId": 22902, "appPackage": 83707, "userAgent": 65303, "message": {"args": "a,b,c,c", "tag": "tag", "logType": "error", "reqPath": "/api/add", "cost": 100, "result": "success", "clientIp": "1.1.1.2", "resultCode": 200}, "clientType": 30186}
    public static final String KAFKA_LOG_CLOUD_ALERTNAME_KEY = "cloud log alert";
    public static final String KAFKA_LOG_CLOUD_APPNAME_KEY = "appName";
    public static final String KAFKA_LOG_CLOUD_TRACEID_KEY = "traceId";
    public static final String KAFKA_LOG_CLOUD_VIN_KEY = "vin";
    public static final String KAFKA_LOG_CLOUD_DEVICEID_KEY = "deviceId";
    // 云端埋点日志告警
    public static final String KAFKA_LOG_CLOUD_EVENT_ALERTNAME_KEY = "cloud event log alert";
    public static final String KAFKA_LOG_CLOUD_EVENT_DEVICEID_KEY = "vehicleModel";
    public static final String KAFKA_LOG_CLOUD_EVENT_OAUTHPROVIDER_ID_KEY = "deviceId";
    public static final String KAFKA_LOG_CLOUD_EVENT_VIN_KEY = "vin";

    public static final String SENTRY_LINK = "https://ivi-sit.faw.cn:39100/organizations/sentry/issues/";
    public static final String SENTRY_LOG_APPLICATION_ID = "data.contexts.app.appIdentifier";
    public static final String SENTRY_LOG_BEHAVIOR_ID = "";
    public static final String SENTRY_LOG_VIN = "";
    public static final String SENTRY_LOG_UID = "";

    // AlertManager 配置
    public static String AM_HOST;
    public static int AM_PORT;
    public static Route AM_ROUTE;

    // Mysql 配置
    public static String MYSQL_CONF_FILE;
    public static String MYSQL_RULE_SQL;
    public static String MYSQL_DRIVER_CLASS;
    public static String MYSQL_HOST;
    public static int MYSQL_PORT;
    public static String MYSQL_USER;
    public static String MYSQL_PSWD;
    public static String MYSQL_DB;
    public static String MYSQL_RULE_TABLE;
    public static String MYSQL_DIM_TABLE;
    public static String MYSQL_URL;

    /**
     * 唯一入口类
     * @param t 调用者的类对象
     *
     */
    public static <T> Configuration initApplicationConfig(String[] args, Class<T> t) throws IOException {
        // 初始化 args 配置
        config = ParameterTool.fromArgs(args).getConfiguration();

        // TODO : 自动生成 yarn application name

        initApplicationConfig(config);
        return config;

    }

    public static Configuration initApplicationConfig(Configuration config) throws IOException {
        // 初始化 AppType
        initAppTypeConfig(config);

        // 初始化 Mysql 配置
        initMysqlConfig(config);

        // 初始化 AlertManager 配置
        initAlertManagerConfig(config);

        // 初始化 Kafka 配置
        initKafkaConfig(config);

        log.info("Thread: {}, init ApplicationConfig completed", Thread.currentThread().getId());

        return config;

    }

    private static void initAlertManagerConfig(Configuration config) {
        ConfigOption<Route> routeOption = ConfigOptions.key("am.route").enumType(Route.class).defaultValue(Route.WEBHOOK);
        ConfigOption<String> hostOption = ConfigOptions.key("am.host").stringType().defaultValue("10.6.215.39");
        ConfigOption<Integer> portOption = ConfigOptions.key("am.port").intType().defaultValue(9093);

        AM_ROUTE = config.get(routeOption);
        AM_HOST = config.get(hostOption);
        AM_PORT = config.get(portOption);

        config.set(routeOption, AM_ROUTE);
        config.set(hostOption, AM_HOST);
        config.set(portOption, AM_PORT);

    }

    private static void initMysqlConfig(Configuration config) throws IOException {
        ConfigOption<String> mysqlConfFile = ConfigOptions.key("mysql.conf.file").stringType().defaultValue("alert/mysql.properties");
        ConfigOption<String> mysqlDriverClass = ConfigOptions.key("mysql.driver.class").stringType().defaultValue("com.mysql.jdbc.Driver");   // com.mysql.cj.jdbc.Driver
        MYSQL_CONF_FILE = config.get(mysqlConfFile);
        MYSQL_DRIVER_CLASS = config.get(mysqlDriverClass);

        Configuration musqlConfig = ParameterTool.fromPropertiesFile(Config.class.getClassLoader().getResourceAsStream(MYSQL_CONF_FILE)).getConfiguration();
        ConfigOption<String> userOption = ConfigOptions.key("user").stringType().noDefaultValue();
        ConfigOption<String> pswdOption = ConfigOptions.key("pswd").stringType().noDefaultValue();
        ConfigOption<String> hostOption = ConfigOptions.key("host").stringType().noDefaultValue();
        ConfigOption<Integer> portOption = ConfigOptions.key("port").intType().noDefaultValue();
        ConfigOption<String> dbOption = ConfigOptions.key("db").stringType().noDefaultValue();
        ConfigOption<String> ruleSqlOption = ConfigOptions.key("rule.sql").stringType().noDefaultValue();
        MYSQL_USER = musqlConfig.getString(userOption);
        MYSQL_PSWD = musqlConfig.getString(pswdOption);
        MYSQL_HOST = musqlConfig.getString(hostOption);
        MYSQL_PORT = musqlConfig.getInteger(portOption);
        MYSQL_RULE_SQL = musqlConfig.getString(ruleSqlOption);
        MYSQL_URL =  String.format("jdbc:mysql://%s:%d/%s", musqlConfig.getString(hostOption), musqlConfig.getInteger(portOption), musqlConfig.getString(dbOption));
        musqlConfig.setString("mysql.url", MYSQL_URL);
        config.addAll(musqlConfig);
    }

    private static void initAppTypeConfig(Configuration config) {
        ConfigOption<AppType> option = ConfigOptions.key("app.type").enumType(AppType.class).noDefaultValue();
        log.info("init APP_TYPE: {}", config.get(option));
        APP_TYPE = config.get(option);
    }


    private static void initKafkaConfig(Configuration config) throws IOException {

        /*Configuration kafkaConfig = ParameterTool.fromPropertiesFile(Config.class.getClassLoader().getResourceAsStream(KAFKA_CONSUMER_CONFIG_FILE)).getConfiguration();
        config.addAll(kafkaConfig);*/

        ConfigOption<String> sinkTopic = ConfigOptions.key("kafka.sink.topic").stringType().defaultValue("sentry-sink-topic-test");
        ConfigOption<String> sinkConf = ConfigOptions.key("kafka.sink.conf.file").stringType().defaultValue("alert/producer.properties");
        ConfigOption<String> sourceConf = ConfigOptions.key("kafka.source.conf.file").stringType().defaultValue("alert/kafka.properties");

        KAFKA_PRODUCER_TOPIC = config.get(sinkTopic);
        KAFKA_PRODUCER_CONFIG_FILE = config.get(sinkConf);
        KAFKA_CONSUMER_CONFIG_FILE = config.get(sourceConf);

        switch (APP_TYPE){

            case CLIENT_EVENT:  // 车机端埋点
            case SENTRY:  // 车机端埋点
                KAFKA_CONSUMER_TOPIC_LIST = Collections.singletonList("data-monitor-ivi");
                break;

            case CLOUD_LOG:   // 云端微服务日志告
                KAFKA_CONSUMER_TOPIC_LIST = Collections.singletonList("data-monitor-cloud");
                break;

            case CLOUD_EVENT: // 云端埋点日志
                KAFKA_CONSUMER_TOPIC_LIST = Collections.singletonList("cloud_cp_data");
                break;

            // case SENTRY: // 云端埋点日志
            //     sourceTopic = "events";
            //     break;

            default: throw new RuntimeException("unknown route, route can not null ");

        }


    }


}
